package com.shujia.interfacer;


import java.io.BufferedWriter;
import java.io.FileWriter;
import java.io.IOException;
import java.util.*;

public class DataSet<T> {
    List<T> list;

    public DataSet(ArrayList<T> list) {
        this.list = list;
    }

    /**
     * 对集合中的元素进行处理
     */
    public <OUT> DataSet<OUT> map(Funcation<T, OUT> funcation) {
        ArrayList<OUT> integers = new ArrayList<>();
        for (T integer : list) {

            //将具体逻辑交给调用者
            OUT add = funcation.map(integer);

            integers.add(add);
        }

        return new DataSet<>(integers);

    }

    /**
     * 对数据进行过滤
     *
     * @param funaction
     * @return
     */

    public DataSet<T> filter(FilterFunaction<T> funaction) {
        ArrayList<T> integers = new ArrayList<>();
        for (T integer : list) {

            //将判断逻辑抽象成接口
            boolean flag = funaction.filter(integer);

            if (flag) {
                integers.add(integer);
            }
        }

        list = integers;

        return this;
    }


    /**
     * 指定聚合的key和聚合的value进行聚合操作
     *
     * @param keySelect
     * @param valueSelect
     * @return
     */

    public <KEY> DataSet<KeyValue> reduce(KeySelect<T, KEY> keySelect, ValueSelect<T> valueSelect) {

        HashMap<KEY, Long> hashMap = new HashMap<>();

        for (T t : list) {

            //分组的key通过接口暴露给用户
            KEY sid = keySelect.select(t);
            Long s = valueSelect.select(t);

            Long integer = hashMap.get(sid);
            if (integer == null) {
                hashMap.put(sid, s);
            } else {
                hashMap.put(sid, integer + s);
            }

        }

        ArrayList<KeyValue> list = new ArrayList<>();

        for (Map.Entry<KEY, Long> entry : hashMap.entrySet()) {
            KeyValue<KEY, Long> keyValue = new KeyValue<>(entry.getKey(), entry.getValue());
            list.add(keyValue);
        }

        return new DataSet<>(list);

    }


    //打印结果
    public void print() {
        for (T t : list) {
            System.out.println(t);
        }
    }


    /**
     * 保存数据
     */
    public void save(String path) {
        BufferedWriter bufferedWriter = null;
        try {
            bufferedWriter = new BufferedWriter(new FileWriter(path));
            for (T t : list) {


                bufferedWriter.write(t.toString());
                bufferedWriter.newLine();

            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (bufferedWriter != null) {
                try {
                    bufferedWriter.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

    }


    /**
     * 取topN
     *
     * @param top
     * @return
     */

    public DataSet<T> take(int top) {
        list = list.subList(0, top);

        return this;
    }

    public DataSet<T> sort(Comparator<T> comparator) {
        list.sort(comparator);
        return this;
    }

    public <KEY, R, OUT> DataSet<OUT> join(DataSet<R> ds, KeySelect<T, KEY> key1, KeySelect<R, KEY> key2, ProcessFunaction<T, R, OUT> processFunaction) {


        ArrayList<OUT> outs = new ArrayList<>();

        for (R r : ds.list) {
            for (T t : list) {
                if (key1.select(t).equals(key2.select(r))) {
                    OUT out = processFunaction.process(t, r);
                    outs.add(out);
                }
            }

        }

        return new DataSet<>(outs);

    }


}
